#include "Spark/Net/EventLoop.h"
#include "Spark/Utils/Logger.h"
#include "Spark/Utils/Funcs.h"
#include "ZKClient.h"

void watcherFn(int type,
               int state,
               const std::string& path,
               ZKClient::ptr client)
{
    LOG_INFO << "In Zookeeper watcher_fn:"
             << "\ntype = " << type << "\nstate = " << state
             << "\npath = " << path << "\nclient = " << client.get()
             << "\nEventLoop = " << EventLoop::getEventLoopOfThisThread();
    if (state == ZOO_CONNECTED_STATE)
    {
        std::string new_val(256, 0);
        int ret = client->Create(
            "/zktest", "", new_val, &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL);
        if (ret == ZOK)
        {
            LOG_INFO << "[" << new_val << "]";
        }
        else
        {
            LOG_INFO << "create error " << ret;
        }

        ret = client->Create("/zktest",
                             "",
                             new_val,
                             &ZOO_OPEN_ACL_UNSAFE,
                             ZOO_SEQUENCE | ZOO_EPHEMERAL);
        if (ret == ZOK)
        {
            LOG_INFO << "create [" << new_val << "]";
        }
        else
        {
            LOG_INFO << "create error " << ret;
        }

        ret = client->Get("/hello", new_val, true);
        if (ret == ZOK)
        {
            LOG_INFO << "get [" << new_val << "]";
        }
        else
        {
            LOG_INFO << "get error " << ret;
        }

        ret = client->Create(
            "/hello", "", new_val, &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL);
        if (ret == ZOK)
        {
            LOG_INFO << "get [" << new_val << "]";
        }
        else
        {
            LOG_INFO << "get error " << ret;
        }

        ret = client->Set("/hello", "world");
        if (ret == ZOK)
        {
            LOG_INFO << "set [" << new_val << "]";
        }
        else
        {
            LOG_INFO << "set error " << ret;
        }

        ret = client->Delete("/hello");
        if (ret == ZOK)
        {
            LOG_INFO << "del [" << new_val << "]";
        }
        else
        {
            LOG_INFO << "del error " << ret;
        }
    }
    else if (state == ZOO_EXPIRED_SESSION_STATE)
    {
        client->Reconnect();
    }
    else
    {
        LOG_ERROR << "Unkown state";
    }
}

int main(int argc, char** argv)
{
    if (argc == 1 && argc >= 3)
    {
        LOG_ERROR << "Usage: " << argv[0] << " [zkip:port]";
        return 1;
    }
    EventLoop loop;
    ZKClient::ptr zkcli(new ZKClient);
    zkcli->Init(argv[1], 3000, watcherFn);
    auto id = loop.runEvery(3_s, [&zkcli] {
        LOG_INFO << "zkcli GetState: " << zkcli->GetState();
    });
    loop.runAfter(30_s, [&id, &loop] {
        LOG_INFO << "Exit...";
        loop.disableTimer(id);
    });
    loop.runAfter(31_s, [&zkcli, &loop] {
        LOG_INFO << "zkcli close";
        zkcli->Close();
        loop.quit();
    });
    loop.loop();
    return 0;
}